﻿using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Reactive.Concurrency;
using System.Reactive.Disposables;
using System.Reactive.Linq;
using System.Text;
using Utils.Common;

namespace Utils
{
    public static class Helpers
    {
        public static IObservable<byte> ToObservable(this FileStream source, int buffersize, IScheduler scheduler)
        {
            var bytes = Observable.Create<byte>(o =>
            {
                var initialState = new StreamReaderState(source, buffersize);
                var currentStateSubscription = new SerialDisposable();
                Action<StreamReaderState, Action<StreamReaderState>> iterator =
                (state, self) =>
                currentStateSubscription.Disposable = state.ReadNext()
                .Subscribe(
                bytesRead =>
                {
                    for (int i = 0; i < bytesRead; i++)
                    {
                        o.OnNext(state.Buffer[i]);
                    }
                    if (bytesRead > 0)
                        self(state);
                    else
                        o.OnCompleted();
                },
                o.OnError);
                var scheduledWork = scheduler.Schedule(initialState, iterator);
                return new CompositeDisposable(currentStateSubscription, scheduledWork);
            });
            return Observable.Using(() => source, _ => bytes);
        }


    }
}
